package com.ccnu.demojava.flink;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.source.TransactionSource;

public class FraudDetectionJob {

    public static void main(String[] args) throws Exception {
        // 设置执行环境,是您为 Job 设置属性、创建源并最终触发 Job 执行的方式。StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 创建数据源，从外部系统（如 Apache Kafka、Rabbit MQ 或 Apache Pulsar）提取到 Flink 作业中
        DataStream<Transaction> transactions = env
                .addSource(new TransactionSource())
                .name("transactions");
        // 对事件进行分区和检测欺诈
        DataStream<Alert> alerts = transactions
                .keyBy(Transaction::getAccountId)
                .process(new FraudDetector())
                .name("fraud-detector");
        // 输出结果
        alerts.addSink(new AlertSink())
              .name("send-alerts");
        // 执行job
        env.execute("Fraud Detection");
    }
}
